package com.jf.net.adapter

import kotlinx.coroutines.CancellableContinuation
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.flow.take
import kotlinx.coroutines.suspendCancellableCoroutine
import retrofit2.Call
import retrofit2.CallAdapter
import retrofit2.Callback
import retrofit2.HttpException
import retrofit2.Response
import kotlin.coroutines.intrinsics.COROUTINE_SUSPENDED
import kotlin.coroutines.intrinsics.intercepted
import kotlin.coroutines.intrinsics.suspendCoroutineUninterceptedOrReturn
import kotlin.coroutines.resume
import kotlin.coroutines.resumeWithException

/**
 * BaseFlowCallAdapter
 * @author JackyFlame
 * @time 2023/10/23 13:26
 * @description CallAdapter for Flow
 */
abstract class BaseFlowCallAdapter<R, T> : CallAdapter<R, Flow<T>> {

    fun <R> bodyFlow(call: Call<R>): Flow<R> = toFlow(call) { response, continuation->
        //http响应[200..300)，恢复执行，并返回响应体
        response.body()?.also {
            continuation.resume(it)
        } ?: let {
            continuation.resumeWithException(HttpException(response))
        }
    }

    fun <R> asyncBodyFlow(call: Call<R>): Flow<R> = asyncToFlow(call) { response, continuation->
        //http响应[200..300)，恢复执行并返回响应体
        response.body()?.also {
            continuation.resume(it)
        } ?: let {
            continuation.resumeWithException(HttpException(response))
        }
    }

    fun <T> responseFlow(call: Call<T>): Flow<Response<T>> = toFlow(call) { response, continuation->
        continuation.resume(response)
    }

    fun <R> asyncResponseFlow(call: Call<R>): Flow<Response<R>> = asyncToFlow(call) { response, continuation->
        continuation.resume(response)
    }.take(1)

    /**
     * 异步读取数据并返回flow
     * */
    private fun <X, Y> toFlow(
        call: Call<X>,
        block: (Response<X>, CancellableContinuation<Y>) -> Unit
    ) : Flow<Y> = flow {
        cancellableCoroutine(call) { realCall, continuation ->
            try {
                //执行call.execute()
                realCall.execute().also {
                    //恢复执行，并返回Response
                    if (it.isSuccessful) {
                        block(it, continuation)
                    } else {
                        continuation.resumeWithException(HttpException(it))
                    }
                }
            } catch (e: Exception) {
                //捕获异常，恢复执行，并返回异常
                continuation.resumeWithException(e)
            }
        }.let { response ->
            //通过flow发射Response
            emit(response)
        }
    }

    /**
     * 同步读取并返回flow
     * */
    private fun <X, Y> asyncToFlow(
        call: Call<X>,
        block: (Response<X>, CancellableContinuation<Y>) -> Unit
    ): Flow<Y> = flow {
        try {
            cancellableCoroutine(call) { req, continuation ->
                //参考retrofit2: call.awaitResponse()
                req.enqueue(object : Callback<X> {
                    override fun onResponse(call: Call<X>, response: Response<X>) {
                        if (response.isSuccessful) {
                            //http响应[200..300)，恢复执行并返回响应体
                            block(response, continuation)
                        }else{
                            //其他http响应，恢复执行并抛出http异常
                            continuation.resumeWithException(HttpException(response))
                        }
                    }
                    override fun onFailure(call: Call<X>, t: Throwable) {
                        //其他http响应，恢复执行并抛出http异常
                        continuation.resumeWithException(t)
                    }
                })
            }.let {
                //通过flow发射响应体
                emit(it)
            }
        } catch (e: Exception) {
            //参考retrofit2: Exception.suspendAndThrow()
            suspendCoroutineUninterceptedOrReturn<Nothing> { continuation ->
                Dispatchers.Default.dispatch(continuation.context) {
                    //特殊case处理，确保抛出异常前挂起
                    continuation.intercepted().resumeWithException(e)
                }
                COROUTINE_SUSPENDED
            }
        }
    }

    /**
     * 检测并返回call，若协程重试则通过clone重置call引用
     * */
    private fun <X> withCall(call: Call<X>): Call<X> {
        return if (call.isExecuted) {
            SELog.d(SEConst.TAG_NET, "BaseFlowCallAdapter", " call clone with isExecuted")
            call.clone()
        } else {
            call
        }
    }

    /**
     * 创建统一suspendCancellableCoroutine，并检测call
     * */
    private suspend fun <X, Y> cancellableCoroutine(
        call: Call<X>,
        block: (Call<X>, CancellableContinuation<Y>) -> Unit
    ): Y = suspendCancellableCoroutine { continuation ->
        withCall(call).also { realCall ->
            //协程取消时，调用Call.cancel()取消call
            continuation.invokeOnCancellation {
                realCall.cancel()
            }
            block(realCall, continuation)
        }
    }

}